package com.atguigu.gmall.realtime.utils;

/**
 * @Description: kafka消费者工具类
 * @Author: tiancy
 * @Create: 2021/11/24
 */

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.time.Duration;
import java.util.Properties;


public class MyKafkaUtil {
    private static final String KAFKA_SERVER = "hadoop102:9092,hadoop103:9092,hadoop104:9092";

    /**
     * 获取kafka消费者对象
     *
     * @param topic
     * @param groupId
     * @return FlinkKafkaConsumer
     */
    public static FlinkKafkaConsumer<String> getKafkaSource(String topic, String groupId) {
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVER);
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);

        FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), props);
        return kafkaSource;
    }

    /**
     * 获取kafka生产者对象.这里是Flink已经帮我们实现好的,需要我们指定Kafka集群位置,主题和 序列化器.
     * 这个方法在使用时,会存在一个小问题,就是必须指定一个主题名称.并且在Flink中 它指定的这个SimpleStringSchema 序列化器,只是声明了最少一次的语义,没有保证写入时的幂等性.可能会造成数据重复.
     *
     * @param topic
     * @return
     */
   /* public static FlinkKafkaProducer<String> getKafkaSink(String topic) {
        return new FlinkKafkaProducer<String>(KAFKA_SERVER, topic, new SimpleStringSchema()); // 点这个对象`FlinkKafkaProducer`的构造器来找 : 一次性语义==> Semantic.AT_LEAST_ONCE
    }*/
    public static FlinkKafkaProducer<String> getKafkaSink(String topic) {
        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVER);
        props.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 15 * 60 * 1000 + "");

        return new FlinkKafkaProducer<String>("default_topic", new KafkaSerializationSchema<String>() {

            @Override
            public ProducerRecord<byte[], byte[]> serialize(String str, @Nullable Long timestamp) {
                return new ProducerRecord<byte[], byte[]>(topic, str.getBytes());
            }
        }, props, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }


    /**
     * 声明一个通用的获取 FlinkKafkaProducer的生产者对象,通过这个对象像下游写数据,在生产数据时需要指定的 `topic`是包含在当前行数据中的.
     * 需要从当前行数据中取出来,再指定主题.
     *
     * @param kafkaSerializationSchema
     * @param <T>
     * @return
     */
    // <T> FlinkKafkaProducer<T> 前面的 T声明当前静态方法为一个泛型方法,后面T指定的是处理数据的泛型.
    public static <T> FlinkKafkaProducer<T> getKafkaByCustomSchema(KafkaSerializationSchema<T> kafkaSerializationSchema) {
        Properties prop = new Properties();
        prop.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVER);
        // String.valueOf(Duration.ofSeconds(5))
        prop.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 15 * 60 * 1000 + "");
        /**
         * @param String defaultTopic : 如果没有指定当前数据发往的主题,则发往默认的主题中.
         * @param KafkaSerializationSchema<IN> serializationSchema : 声明当前数据的序列哈过程,再往下游写入.
         * @param Properties producerConfig : 指定Kafka配置项中的具体值.
         * @param FlinkKafkaProducer.Semantic semantic : 什么语义化.
         */
        Duration.ofSeconds(5);
        return new FlinkKafkaProducer<T>(
                "default_topic", // 默认主题、处理数据时的自定义序列化器、Kafka生产者必须的属性、指定精准一次性语义的级别.
                kafkaSerializationSchema,
                prop,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE); // 在当前类中FlinkKafkaProducer会有 `beginTransaction`这个方法,是拿Semantic的取值来看的,如果是EXACTLY_ONCE,则会帮我们开启一个提交的事务.
    }

    /**
     * 使用Flink SQL时,创建动态表需要指定`获取kafka连接属性字符串`
     * @param topic : 传入主题名称
     * @param groupId : 传入消费者组
     * @return
     */
    public static String getDDL(String topic, String groupId) {
        return " 'connector' = 'kafka'," +
                "  'topic' = '" + topic + "'," +
                "  'properties.bootstrap.servers' = '" + KAFKA_SERVER + "'," +
                "  'properties.group.id' = '" + groupId + "'," +
                "  'scan.startup.mode' = 'latest-offset'," +
                "  'format' = 'json'";
    }
}
